#include <assert.h>
#include <err.h>
#include <stdlib.h>
#include <unistd.h>

#include "pipeline.h"

#if _PIPELINE_DEBUG != 0
#include <stdio.h>
#endif
#include <stdio.h>

/** <DEFS> **/

#define ERR_FORK  "pipeline: fork() error"
#define ERR_PIPE  "pipeline: pipe() error"


/** <PROTO> **/

/* Assert-check parameters for pipeline(); Try to fix some client mistakes */
static inline void pl_check(const iface_t* main_bw_if,
                            stage_spec_t specs[],
                            unsigned int n_stages);

/* Open pipes according to parameters; update interfaces with fds */
void open_pipes(iface_t* fw_if, iface_t* bw_if,
                unsigned int n_pipes, fw_pipes_dirs_t fwp_dirs);

/* Close every pipe fd in the interface */
void close_pipes(iface_t* iface);

/* Close every pipe fd in the interface except the ones at given index */
void close_pipes_except(iface_t* iface, unsigned int except);

/* Swap (*i1) and (*i2) pointers */
static inline void iface_swap(iface_t** i1, iface_t** i2);

/** </PROTO> **/

/** <DEFS> **/


/** <IMPLEMENTATION> **/

void pipeline(  iface_t*       main_bw_if,
                stage_spec_t   specs[],
                unsigned int   n_stages  ) {

  /* Initialization of structures to hold fds for pipes */
  iface_t if_buf[3];
  iface_t* const cur_fw_if  = &(if_buf[2]);
  iface_t*       cur_bw_if  = NULL;
  iface_t*       nxt_bw_if  = &(if_buf[1]);
  unsigned int s;

  #if _PIPELINE_DEBUG != 0
  fprintf(stderr, "<pl>: Checking parameters ... ");
  #endif

  /* Pre-checks and initializations */
  pl_check(main_bw_if, specs, n_stages);

  #if _PIPELINE_DEBUG != 0
  puts("ok");
  #endif

  for (s = 0; s < n_stages; ++s) {
    stage_spec_t* sspec;
    unsigned int p;

    #if _PIPELINE_DEBUG != 0
    fprintf(stderr, "<pl>: Stage %d:\n", s);
    #endif

    sspec = &(specs[s]);
    /* Last bw_if is main_bw_if */
    if (s == (n_stages - 1)) {
      nxt_bw_if = main_bw_if;
    }

    open_pipes(cur_fw_if, nxt_bw_if, sspec->n_pipes, sspec->fwp_dirs);

    for (p = 0; p < sspec->n_procs; ++p) {
      pid_t pid;

      pid = fork();
      if (pid < 0) {
        err(EXIT_FAILURE, ERR_FORK);
      }

      if (pid == 0) {
        /* Child */
        #if _PIPELINE_DEBUG != 0
        fprintf(stderr, "<%d.%d (%d)> : started\n", s, p, getpid());
        #endif

        /* Close next backward interface */
        close_pipes(nxt_bw_if);
        /* If forward pipes are private in this stage, close unneeded ends */
        if (sspec->fwp_shr == PIPES_PVT) {
          close_pipes_except(cur_fw_if, p);
        }

        /* Call the task */
        sspec->task(p, cur_bw_if, cur_fw_if);

        #if _PIPELINE_DEBUG != 0
        fprintf(stderr, "<%d.%d (%d)> : task completed\n", s, p, getpid());
        #endif
        exit(EXIT_SUCCESS);
      }/* END OF: Child */

      /* Parent */
      #if _PIPELINE_DEBUG != 0
      fprintf(stderr, "<pl>: \t%d.%d (%d) spawned\n", s, p, pid);
      #endif
    } /* END OF: foreach proc */

    /* Close pipes for current stage (are now unneeded) */
    close_pipes(cur_fw_if);
    if (cur_bw_if != NULL) {
      close_pipes(cur_bw_if);
    }
    else {
      cur_bw_if = &(if_buf[0]); /* delayed init: at first stage was NULL */
    }

    /* Update interface pointers */
    iface_swap(&cur_bw_if, &nxt_bw_if);

  } /* END OF: foreach stage */
} /* END OF: pipeline() */


/** </PROTO_IMPLEMENTATION> **/

static inline void pl_check(const iface_t* main_bw_if,
                            stage_spec_t specs[],
                            unsigned int n_stages) {
  unsigned int i;
  assert( (main_bw_if != NULL) &&
          (specs != NULL)      &&
          IN_RANGE(n_stages, 1, MAX_STAGES) );
  for (i = 0; i < n_stages; ++i) {
    register stage_spec_t* s; /* shortcut to current stage specification */
    s = &(specs[i]);
    assert( (s != NULL)                                   &&
            IN_RANGE(s->n_procs, 1, MAX_PROCS_PER_STAGE)  &&
            IN_RANGE(s->n_pipes, 1, MAX_PIPES_PER_STAGE)  &&
            (s->task != NULL)  );
    /* For private fw pipes, n_pipes and n_procs have to match */
    if (s->n_pipes != s->n_procs) s->fwp_shr = PIPES_SHA;
  }
}

void open_pipes(iface_t* fw_if, iface_t* bw_if,
                unsigned int n_pipes, fw_pipes_dirs_t fwp_dirs) {
  unsigned int i;

  for (i = 0; i < n_pipes; ++i) {
    int fd[2];
    if (pipe(fd) < 0) err(EXIT_FAILURE, ERR_PIPE);
    fw_if->w_pipes[i] = fd[1];
    bw_if->r_pipes[i] = fd[0];

    if (fwp_dirs == PIPES_BI) {
      /* Bi-directional */
      int fd2[2];
      if (pipe(fd2) < 0) err(EXIT_FAILURE, ERR_PIPE);
      fw_if->r_pipes[i] = fd2[0];
      bw_if->w_pipes[i] = fd2[1];
    }
    else {
      /* Mono-directional */
      fw_if->r_pipes[i] = PIPE_CLOSED;
      bw_if->w_pipes[i] = PIPE_CLOSED;
    }
  }
  /* Update pipe count */
  fw_if->count = n_pipes;
  bw_if->count = n_pipes;
}

void close_pipes(iface_t* iface) {
  unsigned int i;
  for(i = 0; i < iface->count; ++i) {
    close(iface->r_pipes[i]);
    iface->r_pipes[i] = PIPE_CLOSED;
    close(iface->w_pipes[i]);
    iface->w_pipes[i] = PIPE_CLOSED;
  }
}

void close_pipes_except(iface_t* iface, unsigned int except) {
  unsigned int i;
  for(i = 0; i < iface->count; ++i) {
    if (i != except) {
      close(iface->r_pipes[i]);
      iface->r_pipes[i] = PIPE_CLOSED;
      close(iface->w_pipes[i]);
      iface->w_pipes[i] = PIPE_CLOSED;
    }
  }
}

static inline void iface_swap(iface_t** i1, iface_t** i2) {
  iface_t* tmp = *i1;
  *i1 = *i2;
  *i2 = tmp;
}


/** </PROTO_IMPLEMENTATION> **/

/** </IMPLEMENTATION> **/

